package org.example.rxjava.trans;

import io.reactivex.*;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.BiConsumer;
import io.reactivex.functions.BiFunction;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

public class Main {

    //1.1 map
    private static void map() {
        Observable<Float> map = Observable.just("3.14")
                .map(new Function<String, Float>() {
                    @Override
                    public Float apply(String s) throws Exception {
                        return Float.parseFloat(s);
                    }
                });

        map.subscribe(new Observer<Float>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Float aFloat) {
                System.out.println(aFloat);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    //1.2 floatMap
    private static void flatMap() {
        ListWrapper listWrapper1 = new ListWrapper();
        ListWrapper listWrapper2 = new ListWrapper();
        ListWrapper listWrapper3 = new ListWrapper();
        Observable<Integer> objectObservable = Observable.just(listWrapper1, listWrapper2, listWrapper3).flatMap(new Function<ListWrapper, ObservableSource<Integer>>() {
            @Override
            public ObservableSource<Integer> apply(ListWrapper listWrapper) throws Exception {
                return Observable.fromIterable(listWrapper.list);
            }
        });

        objectObservable.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Integer integer) {
                System.out.println(integer);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });

    }

    //1.3 concatMap  order
    private static void contractMap() {
        ListWrapper listWrapper1 = new ListWrapper();
        ListWrapper listWrapper2 = new ListWrapper();
        ListWrapper listWrapper3 = new ListWrapper();
        Observable<Object> objectObservable = Observable.just(listWrapper1, listWrapper2, listWrapper3)
                .concatMap(new Function<ListWrapper, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply(ListWrapper listWrapper) throws Exception {
                        return Observable.fromIterable(listWrapper.list);
                    }
                });

        objectObservable.subscribe(new Observer<Object>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Object o) {
                System.out.println(o);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    //1.4 buffer
    private static void buffer() {
        Observable.just(1, 2, 3, 4, 5)
                .buffer(3, 2)
                .subscribe(new Observer<List<Integer>>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(List<Integer> integers) {
                        System.out.println(integers);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //2.1 concat
    private static void concat() {
        Observable.concat(Observable.just("Hello Java"), Observable.just("Hello Android"))
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        System.out.println(s);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //2.2 merge
    private static void merge() {
        Observable<Integer> just = Observable.just(1, 2, 3);
        Observable<Integer> just1 = Observable.just(9, 8, 7);
        Observable.merge(just, just1)
                .subscribe(new Observer<Integer>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(Integer integer) {
                        System.out.println(integer);
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //2.3 startWith
    private static void startWith() {
        Observable.just("Java")
                .startWith("Hello")
                .subscribe(new Observer<String>() {
                    @Override
                    public void onSubscribe(Disposable d) {

                    }

                    @Override
                    public void onNext(String s) {
                        System.out.print(s + " ");
                    }

                    @Override
                    public void onError(Throwable e) {

                    }

                    @Override
                    public void onComplete() {

                    }
                });
    }

    //3.1 zip
    private static void zip() {
        Observable<String> just = Observable.just("First", "Second");
        Observable<Integer> just1 = Observable.just(1, 2, 3);
        Observable.zip(just, just1, new BiFunction<String, Integer, String>() {
            @Override
            public String apply(String s, Integer integer) throws Exception {
                return s + integer;
            }
        }).subscribe(new Observer<String>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(String s) {
                System.out.println(s);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    //3.2 combineLatest | when value changed , emitter while emmit new combine value
    private static void combineLatest() {
        Observable<Long> interval = Observable.interval(0, 1, TimeUnit.SECONDS);
        Observable<Long> interval1 = Observable.interval(1, 3, TimeUnit.SECONDS);

        Observable.combineLatest(interval, interval1, new BiFunction<Long, Long, Long>() {
            @Override
            public Long apply(Long aLong, Long aLong2) throws Exception {
                return aLong + aLong2;
            }
        }).subscribe(new Observer<Long>() {
            @Override
            public void onSubscribe(Disposable d) {

            }

            @Override
            public void onNext(Long aLong) {
                System.out.println(aLong);
            }

            @Override
            public void onError(Throwable e) {

            }

            @Override
            public void onComplete() {

            }
        });
    }

    //3.3 reduce
    private static void reduce() {
        Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9)
                .reduce(1, new BiFunction<Integer, Integer, Integer>() {
                    @Override
                    public Integer apply(Integer integer, Integer integer2) throws Exception {
                        return integer + integer2;
                    }
                }).subscribe(new BiConsumer<Integer, Throwable>() {
                    @Override
                    public void accept(Integer integer, Throwable throwable) throws Exception {
                        System.out.println(integer);
                    }
                }).dispose();
    }

    //3.4 Collect
    private static void collect(){
        Observable.just("Hello","Java","Kotlin")
                .collect(new Callable<List<String>>() {
                    @Override
                    public List<String> call() throws Exception {
                        return new ArrayList<>();
                    }
                }, new BiConsumer<List<String>, String>() {
                    @Override
                    public void accept(List<String> strings, String s) throws Exception {
                        strings.add(s);
                    }
                }).subscribe(new Consumer<List<String>>() {
                    @Override
                    public void accept(List<String> strings) throws Exception {
                        System.out.println(strings);
                    }
                }).dispose();
    }

    //3.5 count
    private static void count(){
        Observable.just(1,2,3,4,5)
                .count()
                .subscribe(new Consumer<Long>() {
                    @Override
                    public void accept(Long aLong) throws Exception {
                        System.out.println(aLong);
                    }
                }).dispose();
    }

    public static void main(String[] args) {
//        map();
//        flatMap();
//        contractMap();
//        buffer();
//        concat();
//        merge();
//        startWith();
//        zip();
//        combineLatest();
//        reduce();
//        collect();
        count();
        while (true) {
        }
    }

    static class ListWrapper {
        List<Integer> list = Arrays.asList(1, 1, 1, 1, 1, 1, 1);
    }
}
